package spark.core.java;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;

/**
 * 二次排序
 * 1.实现自定义的key，要实现Order接口，和Serializable接口，在key中实现自己对多列的排序算法
 * 2.将包含文本的RDD,映射成为key为自定义KEY,VALUE为文本的JAVAPAIRRDD
 * 3.使用sortByKey,算子按照自定义的key进行排序
 * 4.再次映射，剔除自定义的key，只保留文本行
 */
public class SecondarySort {

    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName("SortWordCount").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);
        final JavaRDD<String> lines = sc.textFile("datas/secondSort.txt");

        JavaPairRDD<SecondarySortKey,String> pairs = lines.mapToPair(new PairFunction<String, SecondarySortKey, String>() {
            public Tuple2<SecondarySortKey, String> call(String line) throws Exception {
                String[] lineSplited = line.split(" ");
                SecondarySortKey key = new SecondarySortKey(Integer.valueOf(lineSplited[0]), Integer.valueOf(lineSplited[1]));
                System.out.println(line);
                return new Tuple2<SecondarySortKey, String>(key, line);
            }
        });
        JavaPairRDD<SecondarySortKey,String> sortedpairs = pairs.sortByKey();
        JavaRDD<String> sortedLines = sortedpairs.map(
                new Function<Tuple2<SecondarySortKey,String>, String>() {
            public String call(Tuple2<SecondarySortKey, String> v1) throws Exception {
                return v1._2();
            }
        });
        sortedLines.foreach(new VoidFunction<String>() {
            public void call(String s) throws Exception {
                System.out.println(s);
            }
        });
    }
}
